package cn.ljy.kafkastudy.consumer;

import cn.ljy.kafkastudy.Consts;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description:
 * @author: James
 * @create: 2022-03-05 13:52
 */
@Slf4j
@Component
public class KafkaConsumer {

    //kafka的监听器，topic为"zhTest"，消费者组为"zhTestGroup"
    @KafkaListener(topics = Consts.TOPIC_NAME, groupId = "zhTestGroup")
    public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        log.info(value);
        log.info("record:"+record);
        //手动提交offset
        ack.acknowledge();
    }

    private AtomicInteger count=new AtomicInteger(0);
    private int limit=2;

    //配置多个消费组
    @KafkaListener(topics = Consts.TOPIC_NAME,groupId = "zhTestGroup2")
    public void listenTulingGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        log.info(value);
        log.info("record:"+record);
        //手动提交offset
        if(count.getAndAdd(1)<=limit){
            ack.nack(1000);
        }else{
            ack.acknowledge();
            count.set(0);
        }
    }

    //配置多个消费组
    @KafkaListener(topics = Consts.TOPIC_NAME2,groupId = "zhTestGroup2")
    public void listenTopic2Group(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        log.info(value);
        log.info("record:"+record);
        //手动提交offset
        ack.acknowledge();
    }
}